package com.jumei.flume.interceptor;

import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import com.jumei.flume.parser.LogLineParser;
import com.jumei.flume.parser.ParsedLine;
import org.apache.commons.lang.StringUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.interceptor.Interceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by zcpeng on 16/6/30.
 */
public class LogLineInterceptor implements Interceptor {
    private static final Logger logger = LoggerFactory.getLogger(LogLineInterceptor.class);

    /**
     * 配置参数parser，用来配置具体使用的Parser
     */
    public static final String PARSER = "parser";


    protected final LogLineParser logLineParser;

    @Override
    public void initialize() {

    }

    @Override
    public void close() {

    }

    public LogLineInterceptor(LogLineParser logLineParser) {
        this.logLineParser = logLineParser;
    }

    @Override
    public Event intercept(Event event) {
        //对于拦截器的调用，分析了source的源码，发现其主要调用了针对批量event处理的方法，
        // 且部分数据源的一行会解析出多行信息，故这里不再单独实现
        logger.error("called intercept single event method,please check it!");


        ArrayList<Event> events = new ArrayList<>();
        events.add(event);


        Event event1 = this.intercept(events).get(0);

        logger.info(event1.getBody().toString());

        return event1;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        List<Event> outEvents = new ArrayList<Event>(events.size());
        
        for (Event event : events) {
            List<ParsedLine> parsedLines = null;
            String line = null;
            try {
                line = new String(event.getBody());
                parsedLines = logLineParser.parse(line);
            } catch (Exception e) {
                logger.error("parse line error，line=" + line,e);
            }

            if (parsedLines == null) {
                continue;
            }

            
            Map<String, String> o_Header = event.getHeaders();
            
            for (ParsedLine parsedLine : parsedLines) {
                if (parsedLine == null) continue;
                Map<String, String> headers = getHeadersByParsedLine(parsedLine);
                headers.putAll(o_Header);//添加原始信息header
                if (parsedLine.isSingleLine()) {
                    String pLine = parsedLine.getLine();
                    logger.info("json字符串"+pLine);
                    if (pLine != null) {
                        SimpleEvent simpleEvent = new SimpleEvent();
                        simpleEvent.setHeaders(headers);
                        simpleEvent.setBody(parsedLine.getLine().getBytes());
                        outEvents.add(simpleEvent);
                    }
                } else {
                    if (parsedLine.getLines() == null) continue;
                    for (String pLine : parsedLine.getLines()) {
                        if (pLine == null) continue;
                        SimpleEvent simpleEvent = new SimpleEvent();
                        simpleEvent.setHeaders(headers);
                        simpleEvent.setBody(pLine.getBytes());
                        outEvents.add(simpleEvent);
                    }
                }
            }
        }

        logger.info("事件大小"+outEvents.size());
        return outEvents;
    }

    /**
     * 从解析结果的ParsedLine中构造event的headers
     *
     * @param parsedLine
     * @return
     */
    private Map<String, String> getHeadersByParsedLine(ParsedLine parsedLine) {
        Map<String, String> headers = new HashMap<String, String>();
        if (parsedLine == null) return headers;

        headers.put("category", parsedLine.getCategory());
        headers.put("year", parsedLine.getCreateYearString());
        headers.put("yearMonth", parsedLine.getCreateYearMonthString());
        headers.put("date", parsedLine.getCreateDateString());
        headers.put("month", parsedLine.getCreateMonthString());
        headers.put("day", parsedLine.getCreateDayString());
        headers.put("hour", parsedLine.getCreateHourString());
        //添加一个备用的参数
        String standbyParam = parsedLine.getStandbyParam();
        if (StringUtils.isNotBlank(standbyParam)) {
            headers.put("standbyParam", standbyParam);
        }

        long timestamp = System.currentTimeMillis();
        if (parsedLine.getCreateTime() != null) {
            timestamp = parsedLine.getCreateTime().getMillis();
        }
        headers.put("timestamp", "" + timestamp);
        return headers;
    }

    /**
     * Builder which builds new instances of the HostInterceptor.
     */
    public static class Builder implements Interceptor.Builder {
        private String parserClassName;

        @Override
        public Interceptor build() {
            LogLineParser theLogLineParser = null;
            Preconditions.checkArgument(parserClassName != null,
                    "parserClassName was misconfigured");
            ClassLoader classLoader = getClass().getClassLoader();
            Class parserClass = null;
            try {
                parserClass = classLoader.loadClass(parserClassName);
            } catch (ClassNotFoundException e) {
                e.printStackTrace();
            }
            Preconditions.checkArgument(parserClass != null,
                    "parserClass was misconfigured");
            try {
                theLogLineParser = (LogLineParser) parserClass.newInstance();
                theLogLineParser.initRegisterSubParsers();
            } catch (Exception e) {
                logger.error("init logLineParser error!", e);
            }

            return new LogLineInterceptor(theLogLineParser);
        }

        @Override
        public void configure(Context context) {
            parserClassName = context.getString(PARSER);
            ImmutableMap<String, String> contextMap = context.getParameters();
            for (Map.Entry<String, String> entry : contextMap.entrySet()) {
                logger.info(entry.getKey() + "=" + entry.getValue());
            }
            logger.info("parserClassName == " + parserClassName);
            Preconditions.checkArgument(!StringUtils.isEmpty(parserClassName),
                    "Must supply a valid LogLineParser class name string");
        }

    }
}
